package com.tiantang.kafka.partitioner;

import java.util.List;
import java.util.Map;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 
 * @description:功能描述 自定义分区计算分方式，需要实现Partitioner接口
 * @author: liujinkun
 * @date 2018年5月27日
 * @version:1.0
 */
public class MyPartitioner implements Partitioner{
	
	private static Logger logger=LoggerFactory.getLogger(MyPartitioner.class);

	/**
	 * 该方法只会调用一次
	 */
	public void configure(Map<String, ?> configs) {
		logger.info("kafka的配置为：configs={}",configs);
	}

	/**
	 * 计算分区
	 */
	public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
		// 根据该topic获取所有可以访问的分区
		List<PartitionInfo> partitions = cluster.availablePartitionsForTopic(topic);
		logger.info("可访问的分区数量为：size={}",partitions.size());
		for (PartitionInfo info : partitions) {
			logger.info("分区号：partition={}",info.partition());
		}
		logger.info("所有的分区数量为：total={}",cluster.partitionsForTopic(topic).size());
		// 返回分区
		return 0;
	}

	/**
	 * 该方法只会调用一次
	 */
	public void close() {
		logger.info("生产者关闭连接时，自定义分区中打印的日志信息");
	}

}
